/**
 * 
 */
package qy.jalgotrade.event;

import java.time.ZonedDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

import org.apache.commons.lang3.tuple.Pair;

import qy.jalgotrade.utils.CommonUtils;

/**
 * This class is responsible for dispatching events from multiple subjects, synchronizing them if
 * necessary.
 * 
 * <pre>
 * 负责: Event 调度, Subject (事件处理器 / 观察者) 管理, Event 调度处理过程的同步:
 * event-source: 所管理的每个 subject, 采用 PULL 模式, 轮询 subject (subject#dispatch()), 委派其上注册的 event-handler
 * 实现具体的事件处理;
 * main-event-loop: Dispatcher#run(), 当所管理的每个 subject 都到达 EOF 状态时 (subject#eof()), 结束
 * main-event-loop.
 * </pre>
 * 
 * @author c-geo
 *
 */
public class Dispatcher {

	public enum DispatcherPrio {
		// @formatter:off
		FIRST(0),
		// We want to process order events before bar feed events.
		BROKER(1000),
		BAR_FEED(2000),
		SRC_BAR_FEED(3000),
		RESAMPLED_BAR_FEED_I(3001),
		RESAMPLED_BAR_FEED_II(3002),
		RESAMPLED_BAR_FEED_III(3003),
		RESAMPLED_BAR_FEED_IV(3004),
		RESAMPLED_BAR_FEED_V(3005),
		RESAMPLED_BAR_FEED_VI(3006),
		LAST(Integer.MAX_VALUE);
		// @formatter:on

		private int value;

		private DispatcherPrio(int value) {

			this.value = value;
		}

		public int getValue() {

			return value;
		}
	}

	/**
	 * 管理的 Subject 列表
	 */
	private List<Subject> __subjects;

	private boolean __stop;

	private Event __startEvent;

	private Event __idleEvent;

	private ZonedDateTime __currDateTime;

	/**
	 * 
	 */
	public Dispatcher() {

		__subjects = new ArrayList<>();
		__stop = false;
		__startEvent = new Event("dispatcher-start-event");
		__idleEvent = new Event("dispatcher-idle-event");
		__currDateTime = null;
	}

	/**
	 * Returns the current event datetime. It may be None for events from realtime subjects.
	 * 
	 * @return
	 */
	public ZonedDateTime getCurrentDateTime() {

		return __currDateTime;
	}

	public Event getStartEvent() {

		return __startEvent;
	}

	public Event getIdleEvent() {

		return __idleEvent;
	}

	public void stop() {

		__stop = true;
	}

	public List<Subject> getSubjects() {

		return __subjects;
	}

	public void addSubject(Subject subject) {

		// Skip the subject if it was already added.
		if (__subjects.contains(subject)) {
			return;
		}

		// If the subject has no specific dispatch priority put it right at the end.
		if (subject.getDispatchPriority() == DispatcherPrio.LAST) {
			__subjects.add(subject);
		} else {
			// Find the position according to the subject's priority.
			int pos = 0;
			for (Subject e : __subjects) {
				if (e.getDispatchPriority() == DispatcherPrio.LAST
				        || subject.getDispatchPriority().getValue() < e.getDispatchPriority().getValue()) {
					break;
				}
				pos += 1;
			}
			__subjects.add(pos, subject);
		}

		subject.onDispatcherRegistered(this);
	}

	/**
	 * TODO: do Pre-Dispatch.
	 * 
	 * @return
	 * @throws Exception
	 */
	private void __preDispatchSubject(Subject subject, ZonedDateTime currEventDateTime) throws Exception {

		// Pre-Dispatch if the datetime is currEventDateTime of if its a realtime subject.
		ZonedDateTime pDt = subject.peekDateTime();
		if (!subject.eof() && (Arrays.asList(null, currEventDateTime).contains(pDt))) {
			subject.__doPreDispatch();
		}
	}

	/**
	 * Return True if events were dispatched.
	 * 
	 * @throws Exception
	 */
	private boolean __dispatchSubject(Subject subject, ZonedDateTime currEventDateTime) throws Exception {

		boolean ret = false;
		// Dispatch if the datetime is currEventDateTime of if its a realtime subject.
		ZonedDateTime pDt = subject.peekDateTime();
		if (!subject.eof() && (Arrays.asList(null, currEventDateTime).contains(pDt))) {
			ret = subject.__doDispatch() == true;
		}
		return ret;
	}

	/**
	 * Returns a tuple with booleans:<br />
	 * 1: True if all subjects hit eof;<br />
	 * 2: True if at least one subject dispatched events.<br />
	 * <br />
	 * 返回值为一个 tuple (boolean, boolean), 其中:<br />
	 * 1: 当管理的所有 Subject 的 eof() 均为 True时, 返回 True;<br />
	 * 2: 若管理的 Subject 中至少有一个的 dispatch() 返回为 True 时, 返回 True.
	 * 
	 * @return
	 * @throws Exception
	 */
	private Pair<Boolean, Boolean> __dispatch() throws Exception {

		ZonedDateTime smallestDateTime = null;
		boolean eof = true;
		boolean eventsDispatched = false;

		// Scan for the lowest datetime.
		for (Subject e : __subjects) {
			if (!e.eof()) {
				eof = false;
				smallestDateTime = (ZonedDateTime) CommonUtils.safeMin(smallestDateTime, e.peekDateTime()); // TODO:
			}
		}

		// Dispatch realtime subjects and those subjects with the lowest datetime.
		if (!eof) {
			__currDateTime = smallestDateTime;

			// XXX: qy-jalgotrade:
			for (Subject e : __subjects) {
				__preDispatchSubject(e, __currDateTime);
			}

			for (Subject e : __subjects) {
				if (__dispatchSubject(e, smallestDateTime)) {
					eventsDispatched = true;
				}
			}
		}

		return Pair.of(eof, eventsDispatched);
	}

	/**
	 * 一一调用所管理的 Subject 的 start() 后, 开始 dispatch loop:<br />
	 * 1. 若 self.__stop 为 True, 则停止 loop 并结束;<br />
	 * 2. 当管理的所有 Subject 的 eof() 均为 True时, 置 self.__stop 为 True, 停止 loop 并结束:
	 * 
	 */
	public void run() {

		try {
			for (Subject e : __subjects) {
				e.start();
			}

			__startEvent.emit(__startEvent.getName(), Collections.emptyMap());

			while (!__stop) {
				// 主循环:
				Pair<Boolean, Boolean> dispRet = __dispatch();
				boolean eof = dispRet.getLeft();
				boolean eventsDispatched = dispRet.getRight();
				if (eof) {
					__stop = true;
				} else if (!eventsDispatched) {
					__idleEvent.emit(__idleEvent.getName(), Collections.emptyMap());
				}
			}
		} catch (Exception ex) {
			// TODO: handle exception
			ex.printStackTrace();
		} finally {
			// There are no more events.
			__currDateTime = null;

			for (Subject e : __subjects) {
				e.stop();
			}
			for (Subject e : __subjects) {
				e.join();
			}
		}
	}
}
